package com.baishancloud.log.murloc

import cn.hutool.json.JSONUtil
import com.baishancloud.log.common.env.StreamEnv
import com.baishancloud.log.common.sink.SinkUtil
import com.baishancloud.log.common.source.SourceUtil
import com.baishancloud.log.format.scala.Origin
import com.baishancloud.log.murloc.tools.IPInfo
import org.apache.flink.api.java.utils.ParameterTool
import org.apache.flink.api.scala._
import org.apache.flink.streaming.api.scala.DataStream
import org.apache.flink.streaming.api.windowing.assigners.TumblingProcessingTimeWindows
import org.apache.flink.streaming.api.windowing.time.Time

/**
 *
 * @author ziqiang.wang 
 * @date 2021/12/21 12:34
 */
object Murloc extends Serializable {


  def main(args: Array[String]): Unit = {
    val parameterTool: ParameterTool = ParameterTool.fromArgs(args)
    println(s"输入参数：${parameterTool.toMap.toString}")
    paramCheck(parameterTool)
    //首先加载指定文件内容以初始化类内部map表
    IPInfo.init(parameterTool)
    StreamEnv.builder(parameterTool).build()
    val source1: DataStream[String] = SourceUtil.kafkaOnlyValue(parameterTool, ".金华").uid("c66f6ac0-8cfe-4986-910c-cdc60b9e9f00")
    val source2: DataStream[String] = SourceUtil.kafkaOnlyValue(parameterTool, ".喜鹊").uid("ba89a709-eb20-4f3b-860a-d2df6009371e")
    val source: DataStream[String] = source1.union(source2)
    //StreamEnv.getStreamExecutionEnvironment.disableOperatorChaining().setParallelism(1)
    //val source: DataStream[String] = SourceUtil.fileOnlyValue(parameterTool)
    source
      .map(Origin.parse(_).orNull).name("Origin").uid("f9afb8c3-4ade-4477-b4ab-6fa759140412")
      .filter(_ != null).name("!=null").uid("7e21e786-cbab-4840-afcf-ef1f0c881d1f")
      .filter(new SomeFilter(parameterTool)).name("someFilter").uid("7d8cc89f-aeb7-4d0d-9103-541c786bab76")
      .process(new BitRateCal(parameterTool)).name("慢速ip计算").uid("73167285-1ee8-442a-a414-4e184c54718c")
      .keyBy(_.getKey)
      .window(TumblingProcessingTimeWindows.of(Time.minutes(5)))
      .reduce((x, y) => Result(x.getKey, x.slow_count + y.slow_count, x.total_count + y.total_count)).name("winAgg").uid("ff07df5d-e8f3-4968-a9f8-4eaa5a8f6e31")
      .map(JSONUtil.toJsonStr(_)).name("toJsonStr").uid("3feb4f65-4ddc-40da-81b8-ed8be5208172")
      .addSink(SinkUtil.starRocksJsonString(parameterTool)).setParallelism(1).name("starRocks").uid("7015f355-db1a-49ba-9fbc-e6e72f1e6bbf")
    //.print("结果数据：")

    StreamEnv.execute(parameterTool)
  }

  /**
   * 必填参数检查
   */
  private def paramCheck(parameterTool: ParameterTool): Unit = {
    if (!parameterTool.has(domainIds)) {
      throw new RuntimeException(s"未设置 --${domainIds} 参数")
    }
    if (!parameterTool.has(dbFilePath)) {
      throw new RuntimeException(s"未设置 --${dbFilePath} 参数")
    }
  }

}
